Zeabur AI Hub: すべてのAIモデル向け統合APIゲートウェイ

# Install required packages
!pip install openai python-dotenv requests pillow -q

import os
from dotenv import load_dotenv
from openai import OpenAI
import json

# Load environment variables from .env file
load_dotenv()

# Get API key from environment
ZEABUR_API_KEY = os.getenv("ZEABUR_API_KEY")

if not ZEABUR_API_KEY:
    raise ValueError("ZEABUR_API_KEY not found in .env file")

# Available endpoints
ENDPOINTS = {
    "tokyo": "https://hnd1.aihub.zeabur.ai/",
    "san_francisco": "https://sfo1.aihub.zeabur.ai/"
}

# Initialize client (using Tokyo endpoint by default)
client = OpenAI(
    base_url=ENDPOINTS["tokyo"],
    api_key=ZEABUR_API_KEY
)

print("✅ Setup complete! Client initialized with Tokyo endpoint.")

# Basic non-streaming completion
def simple_chat(model="claude-haiku-4-5", message="Hello! Tell me about AI in 20 words."):
    """Send a simple chat message and get a complete response."""
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        max_tokens=500
    )
    return response.choices[0].message.content

# Test with Claude Haiku
print("Using Claude Haiku 4.5:")
print("-" * 50)
result = simple_chat(model="claude-haiku-4-5")
print(result)
print("\n")

# Test with GPT-4o
print("Using GPT-4o:")
print("-" * 50)
result = simple_chat(model="gpt-4o")
print(result)

def streaming_chat(model="claude-haiku-4-5", message="Tell me a short story about a robot in 30 words."):
    """Stream chat responses token by token."""
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        stream=True,
        max_tokens=300
    )
    
    print(f"Streaming response from {model}:")
    print("-" * 50)
    
    full_response = ""
    for chunk in stream:
        content = chunk.choices[0].delta.content or ''
        if content:
            print(content, end='', flush=True)
            full_response += content
    
    print("\n" + "-" * 50)
    return full_response

# Test streaming
response = streaming_chat(
    model="claude-sonnet-4-5"
)

# Test streaming
response = streaming_chat(
    model="gpt-4o"
)

import time
import re

# Provider mapping
PROVIDERS = {
    "claude": "Anthropic",
    "gpt": "OpenAI",
    "gemini": "Google",
    "grok": "xAI",
    "deepseek": "DeepSeek",
    "llama": "Meta",
    "qwen": "Alibaba",
    "kimi": "Moonshot AI"
}

def get_provider_name(model):
    """Get provider name based on model prefix."""
    for prefix, provider in PROVIDERS.items():
        if model.startswith(prefix):
            return provider
    return "Unknown"

def clean_response(content):
    """Remove thinking tags from model responses."""
    if not content:
        return content
    
    original = content
    
    # Remove thinking content for tags: <think>, <thinking>
    for tag in ['think', 'thinking']:
        # Extract content after closing tag if it exists
        closing = f'</{tag}>'
        if closing in content.lower():
            parts = re.split(closing, content, flags=re.IGNORECASE | re.DOTALL)
            content = parts[-1].strip()
        
        # Remove from opening tag onwards if no closing tag
        opening = f'<{tag}>'
        if opening in content.lower():
            content = re.split(opening, content, flags=re.IGNORECASE | re.DOTALL)[0].strip()
    
    return content if content.strip() else original

def compare_models(prompt, models, max_tokens_map=None):
    """Compare response quality and speed across different models."""
    max_tokens_map = max_tokens_map or {}
    results = []
    
    for model in models:
        print(f"Testing {model}...")
        start_time = time.time()
        
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens_map.get(model, 200)
            )
            
            content = clean_response(response.choices[0].message.content)
            
            results.append({
                "model": model,
                "provider": get_provider_name(model),
                "response": content or "[Empty response]",
                "time_seconds": round(time.time() - start_time, 2),
                "tokens": getattr(response.usage, 'total_tokens', "N/A")
            })
            
        except Exception as e:
            results.append({
                "model": model,
                "provider": get_provider_name(model),
                "response": f"Error: {str(e)[:100]}",
                "time_seconds": 0,
                "tokens": 0
            })
    
    return results

# Test configuration
test_prompt = "Tell me a joke in 20 words."

models_to_test = [
    "claude-haiku-4-5",
    "gpt-4o-mini",
    "grok-4-fast-non-reasoning",
    "deepseek-v3.2-exp",
    "llama-3.3-70b",
    "qwen-3-32",
    "kimi-k2-thinking"
]

# Thinking models need more tokens
max_tokens_map = {
    "kimi-k2-thinking": 800,
    "qwen-3-32": 800,
}

# Run comparison
print("Comparing Models Across Providers")
print("=" * 80)
print(f"Prompt: {test_prompt}\n")

results = compare_models(test_prompt, models_to_test, max_tokens_map)

# Display results
for r in results:
    print(f"\n{'=' * 80}")
    print(f"Provider: {r['provider']} | Model: {r['model']}")
    print(f"Time: {r['time_seconds']}s | Tokens: {r['tokens']}")
    print(f"Response: {r['response'][:500]}")

# Summary
print(f"\n{'=' * 80}")
print("SUMMARY")
print("=" * 80)

successful = [r for r in results if not r['response'].startswith('Error') and not r['response'].startswith('[')]

if successful:
    print(f"Successful: {len(successful)}/{len(results)}")
    print(f"Average time: {sum(r['time_seconds'] for r in successful) / len(successful):.2f}s")
    
    fastest = min(successful, key=lambda x: x['time_seconds'])
    slowest = max(successful, key=lambda x: x['time_seconds'])
    
    print(f"Fastest: {fastest['model']} ({fastest['provider']}) - {fastest['time_seconds']}s")
    print(f"Slowest: {slowest['model']} ({slowest['provider']}) - {slowest['time_seconds']}s")
    
    # Provider breakdown
    print(f"\nProviders tested:")
    providers = sorted(set(r['provider'] for r in successful))
    for provider in providers:
        provider_models = [r['model'] for r in successful if r['provider'] == provider]
        print(f"  - {provider}: {', '.join(provider_models)}")
else:
    print("No successful tests.")